from typing import Any, Dict, Iterable, List, Optional, Set

from sqlalchemy import and_, or_, select


def build_conditions(
    query_params: Any,
    model: Any,
    *,
    exclude: Optional[Set[str]] = None,
    text_like_fields: Optional[Set[str]] = None,
    keyword: Optional[str] = None,
    keyword_fields: Optional[Iterable[str]] = None,
) -> List[Any]:
    """通用：根据 Pydantic/query 对象生成 SQLAlchemy 条件列表。

    - 字符串：默认非空使用 ilike；如指定 text_like_fields，则仅这些字段做 ilike，否则等值。
    - 集合类型：转换为 in_ 条件。
    - *_from/*_to：统一归并为区间条件（>=、<=）。
    - 额外支持 keyword + keyword_fields 的全局模糊匹配。
    """
    ex = exclude or {"page", "size"}
    params: Dict[str, Any] = query_params.model_dump(exclude=ex, exclude_none=True)

    conditions: List[Any] = []
    range_map: Dict[str, Dict[str, Any]] = {}

    # 归并区间参数
    for key, value in params.items():
        if key.endswith("_from") or key.endswith("_to"):
            base, bound = (key[:-5], "from") if key.endswith("_from") else (key[:-3], "to")
            range_map.setdefault(base, {})[bound] = value
            continue

        column = getattr(model, key, None)
        if column is None:
            continue

        if isinstance(value, str):
            v = value.strip()
            if not v:
                continue
            if text_like_fields is None or key in text_like_fields:
                conditions.append(column.ilike(f"%{v}%"))
            else:
                conditions.append(column == v)
        elif isinstance(value, (list, tuple, set)):
            seq = [x for x in value if x is not None]
            if seq:
                conditions.append(column.in_(seq))
        else:
            conditions.append(column == value)

    # 输出区间条件
    for base, bounds in range_map.items():
        column = getattr(model, base, None)
        if column is None:
            continue
        if "from" in bounds:
            conditions.append(column >= bounds["from"])
        if "to" in bounds:
            conditions.append(column <= bounds["to"])

    # 全局关键词匹配
    if keyword and keyword_fields:
        kw = keyword.strip()
        if kw:
            ors = []
            for f in keyword_fields:
                col = getattr(model, f, None)
                if col is not None:
                    ors.append(col.ilike(f"%{kw}%"))
            if ors:
                conditions.append(or_(*ors))

    return conditions


def build_select_stmt(
    model: Any,
    query_params: Any,
    *,
    order_by_desc: Optional[str] = None,
    text_like_fields: Optional[Set[str]] = None,
    keyword: Optional[str] = None,
    keyword_fields: Optional[Iterable[str]] = None,
):
    """通用：构造 select(model) 查询语句，附加条件与排序。"""
    stmt = select(model)
    conditions = build_conditions(
        query_params,
        model,
        text_like_fields=text_like_fields,
        keyword=keyword,
        keyword_fields=keyword_fields,
    )
    if conditions:
        stmt = stmt.where(and_(*conditions))
    if order_by_desc:
        col = getattr(model, order_by_desc, None)
        if col is not None:
            stmt = stmt.order_by(col.desc())
    return stmt